{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 波动率策略分析\n",
    "\n",
    "本笔记本实现基于波动率的量化交易策略"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import akshare as ak\n",
    "import matplotlib.pyplot as plt\n",
    "from scipy.stats import zscore\n",
    "\n",
    "# 设置中文显示\n",
    "plt.rcParams['font.sans-serif'] = ['SimHei']\n",
    "plt.rcParams['axes.unicode_minus'] = False"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 数据准备"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_index_data(index_code='000300', start_date='20100101', end_date='20231231'):\n",
    "    \"\"\"\n",
    "    获取指数数据\n",
    "    \n",
    "    Args:\n",
    "        index_code: 指数代码\n",
    "        \n",
    "    Returns:\n",
    "        pd.Series: 收盘价序列\n",
    "    \"\"\"\n",
    "    try:\n",
    "        df = ak.stock_zh_index_hist_csindex(symbol=index_code, start_date=start_date, end_date=end_date)\n",
    "        df['日期'] = pd.to_datetime(df['日期'])\n",
    "        return df.set_index('日期')['收盘']\n",
    "    except Exception as e:\n",
    "        print(f\"获取指数数据失败: {e}\")\n",
    "        return pd.Series()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 波动率计算"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def calculate_volatility(prices, window=20, annualized=True):\n",
    "    \"\"\"\n",
    "    计算历史波动率\n",
    "    \n",
    "    Args:\n",
    "        prices: 价格序列\n",
    "        window: 计算窗口\n",
    "        annualized: 是否年化\n",
    "        \n",
    "    Returns:\n",
    "        pd.Series: 波动率序列\n",
    "    \"\"\"\n",
    "    try:\n",
    "        returns = np.log(prices/prices.shift(1))\n",
    "        vol = returns.rolling(window).std()\n",
    "        if annualized:\n",
    "            vol = vol * np.sqrt(252)\n",
    "        return vol\n",
    "    except Exception as e:\n",
    "        print(f\"波动率计算失败: {e}\")\n",
    "        return pd.Series()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 波动率突破策略"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def volatility_breakout(prices, vol_window=20, breakout_multiplier=1.5):\n",
    "    \"\"\"\n",
    "    波动率突破策略\n",
    "    \n",
    "    Args:\n",
    "        prices: 价格序列\n",
    "        vol_window: 波动率计算窗口\n",
    "        breakout_multiplier: 突破乘数\n",
    "        \n",
    "    Returns:\n",
    "        pd.Series: 策略信号(1:做多, -1:做空, 0:空仓)\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 计算波动率\n",
    "        vol = calculate_volatility(prices, vol_window, annualized=False)\n",
    "        \n",
    "        # 计算ATR(平均真实波幅)\n",
    "        high_low = prices.rolling(vol_window).max() - prices.rolling(vol_window).min()\n",
    "        atr = high_low.rolling(vol_window).mean()\n",
    "        \n",
    "        # 生成信号\n",
    "        signals = pd.Series(0, index=prices.index)\n",
    "        \n",
    "        # 上轨突破做多\n",
    "        upper_band = prices.shift(1) + breakout_multiplier * atr\n",
    "        signals[prices > upper_band] = 1\n",
    "        \n",
    "        # 下轨突破做空\n",
    "        lower_band = prices.shift(1) - breakout_multiplier * atr\n",
    "        signals[prices < lower_band] = -1\n",
    "        \n",
    "        return signals\n",
    "    except Exception as e:\n",
    "        print(f\"波动率突破策略生成失败: {e}\")\n",
    "        return pd.Series()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 波动率均值回归策略"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def volatility_mean_reversion(prices, vol_window=20, z_threshold=1.0):\n",
    "    \"\"\"\n",
    "    波动率均值回归策略\n",
    "    \n",
    "    Args:\n",
    "        prices: 价格序列\n",
    "        vol_window: 波动率计算窗口\n",
    "        z_threshold: Z-score阈值\n",
    "        \n",
    "    Returns:\n",
    "        pd.Series: 策略信号(1:做多, -1:做空, 0:空仓)\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 计算波动率\n",
    "        vol = calculate_volatility(prices, vol_window, annualized=False)\n",
    "        \n",
    "        # 计算Z-score\n",
    "        z_scores = zscore(vol, nan_policy='omit')\n",
    "        \n",
    "        # 生成信号\n",
    "        signals = pd.Series(0, index=prices.index)\n",
    "        \n",
    "        # 高波动率做空\n",
    "        signals[z_scores > z_threshold] = -1\n",
    "        \n",
    "        # 低波动率做多\n",
    "        signals[z_scores < -z_threshold] = 1\n",
    "        \n",
    "        return signals\n",
    "    except Exception as e:\n",
    "        print(f\"波动率均值回归策略生成失败: {e}\")\n",
    "        return pd.Series()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 策略回测"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def backtest(prices, signals, transaction_cost=0.001):\n",
    "    \"\"\"\n",
    "    策略回测\n",
    "    \n",
    "    Args:\n",
    "        prices: 价格序列\n",
    "        signals: 交易信号\n",
    "        transaction_cost: 交易成本\n",
    "        \n",
    "    Returns:\n",
    "        pd.Series: 策略收益率\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 计算价格变化\n",
    "        price_changes = prices.pct_change()\n",
    "        \n",
    "        # 计算策略收益率\n",
    "        strategy_returns = signals.shift(1) * price_changes\n",
    "        \n",
    "        # 计算交易成本\n",
    "        trades = signals.diff().abs()\n",
    "        strategy_returns = strategy_returns - trades * transaction_cost\n",
    "        \n",
    "        return strategy_returns.dropna()\n",
    "    except Exception as e:\n",
    "        print(f\"回测失败: {e}\")\n",
    "        return pd.Series()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 主分析流程"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 获取数据\n",
    "prices = get_index_data()\n",
    "\n",
    "# 波动率突破策略\n",
    "breakout_signals = volatility_breakout(prices)\n",
    "breakout_returns = backtest(prices, breakout_signals)\n",
    "\n",
    "# 波动率均值回归策略\n",
    "mr_signals = volatility_mean_reversion(prices)\n",
    "mr_returns = backtest(prices, mr_signals)\n",
    "\n",
    "# 计算累计收益\n",
    "cum_breakout = (1 + breakout_returns).cumprod()\n",
    "cum_mr = (1 + mr_returns).cumprod()\n",
    "cum_bh = (1 + prices.pct_change()).cumprod()  # 买入持有\n",
    "\n",
    "# 绘制结果\n",
    "plt.figure(figsize=(12, 6))\n",
    "cum_breakout.plot(label='波动率突破策略')\n",
    "cum_mr.plot(label='波动率均值回归策略')\n",
    "cum_bh.plot(label='买入持有')\n",
    "plt.title('波动率策略表现对比')\n",
    "plt.ylabel('累计收益')\n",
    "plt.legend()\n",
    "plt.grid()\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 策略评价"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def evaluate_performance(returns, rf=0.03/252):\n",
    "    \"\"\"\n",
    "    评估策略表现\n",
    "    \n",
    "    Args:\n",
    "        returns: 日收益率序列\n",
    "        rf: 无风险利率(日)\n",
    "        \n",
    "    Returns:\n",
    "        dict: 包含各项指标\n",
    "    \"\"\"\n",
    "    if returns.empty:\n",
    "        return {}\n",
    "        \n",
    "    try:\n",
    "        # 年化收益率\n",
    "        annual_return = (1 + returns.mean())**252 - 1\n",
    "        \n",
    "        # 年化波动率\n",
    "        annual_vol = returns.std() * np.sqrt(252)\n",
    "        \n",
    "        # 夏普比率\n",
    "        sharpe = (returns.mean() - rf) / returns.std() * np.sqrt(252)\n",
    "        \n",
    "        # 最大回撤\n",
    "        cum_returns = (1 + returns).cumprod()\n",
    "        peak = cum_returns.cummax()\n",
    "        drawdown = (cum_returns - peak) / peak\n",
    "        max_drawdown = drawdown.min()\n",
    "        \n",
    "        # 胜率\n",
    "        win_rate = (returns > 0).mean()\n",
    "        \n",
    "        return {\n",
    "            '年化收益率': annual_return,\n",
    "            '年化波动率': annual_vol,\n",
    "            '夏普比率': sharpe,\n",
    "            '最大回撤': max_drawdown,\n",
    "            '胜率': win_rate\n",
    "        }\n",
    "    except Exception as e:\n",
    "        print(f\"策略评价失败: {e}\")\n",
    "        return {}\n",
    "\n",
    "# 评价波动率突破策略\n",
    "breakout_metrics = evaluate_performance(breakout_returns)\n",
    "print(\"波动率突破策略表现:\")\n",
    "for k, v in breakout_metrics.items():\n",
    "    print(f\"{k}: {v:.4f}\")\n",
    "\n",
    "# 评价波动率均值回归策略\n",
    "mr_metrics = evaluate_performance(mr_returns)\n",
    "print(\"\\n波动率均值回归策略表现:\")\n",
    "for k, v in mr_metrics.items():\n",
    "    print(f\"{k}: {v:.4f}\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}